[SPARK-58093][SQL] Add ASOF JOIN SQL syntax with MATCH_CONDITION#57194
[SPARK-58093][SQL] Add ASOF JOIN SQL syntax with MATCH_CONDITION#57194srielau wants to merge 23 commits into
Conversation
Wire the SQL Reference Spec ASOF JOIN surface into the existing AsOfJoin logical operator, reusing the DataFrame execution path (RewriteAsOfJoin / SortMergeAsOfJoinExec). Adds parser grammar, analysis resolution for USING and match operands, spec error conditions, and parser/SQL integration tests.
…pile errors Update case-class pattern matches for the two new AsOfJoin fields, fix conf.resolver usage, replace expr.preOrder with expr.collect, and drop a duplicate RowOrdering import.
… arity The error helper only accepts the two MATCH_CONDITION operands; drop the unused left/right AttributeSet arguments from the throw site.
Rule.ruleId is a lazy val; overriding it with a non-lazy val fails compilation. The base Rule implementation already derives the same id from the class name, so drop the override entirely.
Wrap long lines and reorder expression imports so the braced import precedes expressions.aggregate.
Store parsed MATCH_CONDITION operands in AsOfMatchCondition so mapExpressions does not corrupt the tuple shape. Resolve operands in ResolveReferences and materialize asOfCondition in ResolveAsOfJoin, including USING column projection.
…enabled Default the new SQL syntax off while under development. AstBuilder rejects ASOF JOIN at parse time when disabled; tests opt in via SQLConf.
…iance doc SQLKeywordSuite requires new lexer keywords to be listed in sql-ref-ansi-compliance.md with correct reserved/non-reserved status.
ASOF is a Spark extension, not SQL standard, so it belongs in ansiNonReserved rather than being reserved under ANSI mode.
…sing Restrict matchComparison to valueExpression so boolean AND is rejected at parse time. Tighten type validation for incompatible operand pairs and prepare struct/tuple ordering for field-wise distance expressions.
…expressions Expose parsed MATCH_CONDITION operand trees to analysis pruning and expression rewriting so functions, interval arithmetic, and ON predicates all resolve before ResolveAsOfJoin materializes the join condition.
…erands Recursively build min_by order expressions for nested structs (leaf flattening) and arrays (ZipWith over element-wise diffs). Use signed comparison distance for orderable non-numeric types like STRING. Add AsOfJoinSQLSuite coverage for nested struct, ARRAY<INT>, and struct tuple operands on the min_by rewrite path.
Place TreePattern before TreePattern._ and keep DataTypeUtils imports adjacent per scalastyle rules.
Regenerate keywords golden files to include ASOF and MATCH_CONDITION from the lexer vocabulary. Update bitwise parse-error goldens after the ASOF JOIN grammar change shifts the reported token for `> >>` syntax.
Add ASOF and MATCH_CONDITION to the SPARK-43119 hardcoded keyword list returned by GetInfo(CLI_ODBC_KEYWORDS).
Reorder ASOF_JOIN_MATCH_CONDITION_* entries before ASSIGNMENT_ARITY_MISMATCH and AS_OF_JOIN in error-conditions.json to satisfy SparkThrowableSuite key ordering (ORDER_MAP_ENTRIES_BY_KEYS).
Add ASOF and MATCH_CONDITION to the hardcoded getSQLKeywords expectation after the new ASOF JOIN grammar keywords were registered.
Parse MATCH_CONDITION comparisons as booleanExpression instead of a dedicated matchComparison rule, which disturbed unrelated expression error reporting (e.g. CHECK constraints). Move ASOF JOIN to a trailing grammar alternative and validate allowed comparison operators in AstBuilder. Update the join_ parse-error golden after the new ASOF alternative changes ANTLR error recovery.
Revert the reported error token for `SELECT 20181117 > >> 2` back to '>>' after removing the ASOF matchComparison grammar rule restored the original ANTLR error recovery behavior.
|
@sarutak Are you still actively working on the sort-merge-as-join feature? I need some extensions to generalize the feature. It woudl be helpful to know if you are having anything in flight yourself, or I can extend upon it. |
SQL ASOF JOIN now requires spark.sql.join.sortMergeAsOfJoin.enabled. Extend SortMergeAsOfJoinExec for multi-column MATCH_CONDITION sort keys, fix LEFT OUTER null right-side projection, and add AsOfJoinSortMergeSQLSuite covering scalar types, struct/array operands, and forward match semantics.
srielau
left a comment
There was a problem hiding this comment.
Code review summary
Overall this is a well-structured PR: parser → AsOfJoin.fromMatchCondition → ResolveAsOfJoin → SortMergeAsOfJoinExec is coherent, feature gating is appropriate, and positive-path coverage in AsOfJoinSortMergeSQLSuite is strong.
No P0 correctness bugs found. A few items below are worth addressing before merge (inline comments have details).
Before merge (recommended):
- Fix array element type check (
sameTypevs==) inbuildOrderExpression - Add analysis negative tests for
ASOF_JOIN_MATCH_CONDITION_*error classes - Add a LEFT OUTER regression test for the
nullRightRow/ NOT NULL catalog column case - Update the PR description — latest commit requires sort-merge for SQL and extends
SortMergeAsOfJoinExec, which contradicts the current "Execution: No changes" bullet
Nice-to-have (can follow up):
- Throw
ASOF_JOIN_MATCH_CONDITION_INVALID_OPERATORfor compoundMATCH_CONDITIONinstead of genericPARSE_SYNTAX_ERROR - Strict
>/<and forward-array execution tests sql-refsyntax documentation and dual-config callout
| rightOperand: Expression, | ||
| operator: MatchComparisonOperator): Expression = { | ||
| (leftOperand.dataType, rightOperand.dataType) match { | ||
| case (ArrayType(leftElem, _), ArrayType(rightElem, _)) if leftElem == rightElem => |
There was a problem hiding this comment.
P1: Use structural type equality for array elements
This gate uses reference equality (leftElem == rightElem), while matchSortExpressions (line 2704) uses leftStruct.sameType(rightStruct) and AsOfJoinValidation.areStructurallyComparableTypes allows structurally comparable ARRAY<STRUCT<...>> operands with different field names.
For such operands, analysis accepts the query but buildOrderExpression falls through to buildLeafOrderExpression on the whole array, which is incorrect for forward (<=/<) joins that rely on element-wise distance.
Suggested fix:
case (ArrayType(leftElem, _), ArrayType(rightElem, _))
if DataTypeUtils.sameType(leftElem, rightElem) =>| |""".stripMargin) | ||
| } | ||
|
|
||
| test("SQL ASOF JOIN requires sort-merge conf") { |
There was a problem hiding this comment.
P1: Add analysis negative tests
AsOfJoinValidation rejects subqueries, aggregates, window functions, non-deterministic expressions, incompatible types, and cross-side operands — but none of these paths are covered by SQL integration tests today (only parser-level = rejection in AsOfJoinSQLSuite).
Consider adding checkError tests for at least:
ASOF_JOIN_MATCH_CONDITION_TABLE_REFERENCE(e.g.MATCH_CONDITION (t.a + r.b >= r.c))ASOF_JOIN_MATCH_CONDITION_INVALID_EXPRESSION(e.g. subquery orrand()in operand)ASOF_JOIN_MATCH_CONDITION_INVALID_TYPE(e.g.INTvsSTRING)
| private val nullRightRow = new GenericInternalRow(rightOutput.length) | ||
| // Materialize an all-null right row as UnsafeRow. GenericInternalRow cannot be | ||
| // passed through identity UnsafeProjection when right columns are NOT NULL. | ||
| private val nullRightRow: InternalRow = { |
There was a problem hiding this comment.
P1: Regression test for nullRightRow fix
This fix addresses a real bug: GenericInternalRow null padding fails UnsafeProjection when right-side catalog columns are NOT NULL. The LEFT OUTER tests use VALUES temp views without NOT NULL constraints, so this path isn't exercised in CI.
Consider a test with an explicit NOT NULL right-side schema (catalog table or CREATE TABLE ... NOT NULL) where an unmatched left row must produce null right columns.
| case GreaterThan(left, right) => (left, GreaterThanOp, right) | ||
| case LessThanOrEqual(left, right) => (left, LessThanOrEqualOp, right) | ||
| case LessThan(left, right) => (left, LessThanOp, right) | ||
| case _ => |
There was a problem hiding this comment.
P2: Dead error class / misleading parse error for compound MATCH_CONDITION
MATCH_CONDITION (t.a >= u.a AND t.b >= u.b) hits this branch and surfaces PARSE_SYNTAX_ERROR with error: ')'. Meanwhile ASOF_JOIN_MATCH_CONDITION_INVALID_OPERATOR is defined in error-conditions.json but never thrown.
Consider rejecting non-singular comparisons here with ASOF_JOIN_MATCH_CONDITION_INVALID_OPERATOR (or a clearer dedicated message).
| Row(Date.valueOf("2026-06-29"), Date.valueOf("2026-06-29")) :: Nil) | ||
| } | ||
|
|
||
| test("INT scalar MATCH_CONDITION") { |
There was a problem hiding this comment.
P2: Test coverage gaps
Execution tests cover >= and <= well but not strict > / < (different tie-breaking semantics). Array tests use backward >= only; forward array joins use distance-based early termination in findBestForwardNearest, which is the more fragile path.
Low priority, but one backward > and one forward <= on ARRAY<INT> would close the gap.
I have been considering adding syntax for AS-OF JOIN, but I haven't started working on it yet. Please go ahead and take the lead on this. |
Use DataTypeUtils.sameType for array MATCH_CONDITION operands, throw ASOF_JOIN_MATCH_CONDITION_INVALID_OPERATOR for invalid/compound operators, and expand SQL integration tests for analysis errors, strict operators, and NOT NULL null-padding.
Validate MATCH_CONDITION operands that reference both join inputs before materializing the comparison, so invalid SQL like (t.col + r.col >= r.col) raises ASOF_JOIN_MATCH_CONDITION_TABLE_REFERENCE instead of an internal unresolved-operator error.
Wire the SQL Reference Spec ASOF JOIN surface into the existing
AsOfJoinlogical operator, reusing the DataFrame execution path (RewriteAsOfJoin/SortMergeAsOfJoinExec). Adds parser grammar, analysis resolution forUSINGandMATCH_CONDITIONoperands, spec error conditions, and parser/SQL integration tests.What changes were proposed in this pull request?
This PR adds SQL syntax for ASOF JOIN with a required
MATCH_CONDITIONclause and optionalON/USING, per the SQL Language Reference spec (Snowflake / CalciteMATCH_CONDITIONfamily).Syntax supported (v1):
where
comparison_operatoris one of>=,>,<=,<.Example:
Implementation layers (new vs reused):
ASOF,MATCH_CONDITIONkeywords;asofJoinType,asofJoinCriteria,matchComparisongrammar ruleswithAsOfJoin()buildsAsOfJoin.fromMatchCondition(...)ResolveAsOfJoinrule: expandsUSINGviaNaturalAndUsingJoinResolution, normalizesMATCH_CONDITIONoperands intoasOfCondition+orderExpression, validates operand expressions/typesAsOfJoinwith deferredmatchComparison/usingColumnsfields; addsMatchComparisonOperatorandAsOfJoin.fromMatchConditionspark.sql.join.sortMergeAsOfJoin.enabled=true(sort-merge only;RewriteAsOfJoin/min_bypath is not used for SQL). ExtendsSortMergeAsOfJoinExecfor multi-column sort keys and complexMATCH_CONDITIONtypesASOF_JOIN_MATCH_CONDITION_INVALID_EXPRESSION,..._INVALID_OPERATOR,..._INVALID_TYPE,..._TABLE_REFERENCE, plusAS_OF_JOIN.SORT_MERGE_REQUIREDwhen sort-merge is disabledExplicitly out of scope (deferred per spec):
TOLERANCE,NEARESTdirection,RIGHT/FULL/CROSS/SEMI/ANTIASOF forms.Related prior art in Spark:
DataFrame.joinAsOf(PySpark/Scala),AsOfJoinlogical node,SortMergeAsOfJoinSuite,DataFrameAsOfJoinSuite.Why are the changes needed?
ASOF (as-of / closest-match) join is a standard temporal/analytics pattern — attaching the nearest preceding quote to each trade, finding the next maintenance window for an alert, etc. Spark already implements the semantics via the DataFrame API, but SQL users must hand-write
LEFT JOIN+ROW_NUMBER+QUALIFYrewrites. Native SQL syntax lowers the barrier and aligns with industry dialects (Snowflake, Calcite/Pinot/DorisMATCH_CONDITIONfamily).Does this PR introduce any user-facing change?
Yes.
Before:
ASOF JOINwas not valid SQL; closest-match joins were only available throughDataFrame.joinAsOf.After: SQL queries may use
ASOF JOIN/LEFT ASOF JOINwithMATCH_CONDITION. Behavior matches the spec: asymmetric left-driven lookup, at most one right row per left row,INNERdrops unmatched left rows,LEFTpreserves them with NULL right-side columns.Operand order inside
MATCH_CONDITIONis immaterial (t.ts >= q.ts≡q.ts <= t.ts); the analyzer normalizes to a canonical left-table-on-left form before planning.How was this patch tested?
Added unit and integration tests; full compile/test suite to run on CI (local build blocked by missing
netty 4.2.16artifacts on the dev Maven proxy after rebasing onto latest master).Parser tests (
PlanParserSuite):INNER/LEFT ASOF JOINwithMATCH_CONDITIONandONUSINGclauseu.a <= t.a)=inMATCH_CONDITION(parse error)SQL integration tests (
AsOfJoinSQLSuite):ON,LEFT ASOFpreserving unmatched rows,USINGequivalence, first-following match with<=Existing coverage reused (unchanged):
DataFrameAsOfJoinSuite,SortMergeAsOfJoinSuite,RewriteAsOfJoinSuiteexercise the shared execution path.Suggested CI commands for reviewers:
Was this patch authored or co-authored using generative AI tooling?
Yes. Generated-by: Cursor (Claude agent), with human review and editing by the PR author.